package com.fuda.trade

import android.util.Log
import io.zenoh.Session
import io.zenoh.exceptions.KeyExprException
import io.zenoh.exceptions.ZenohException
import io.zenoh.keyexpr.KeyExpr
import io.zenoh.publication.CongestionControl
import io.zenoh.query.Reply
import io.zenoh.sample.Sample
import io.zenoh.selector.Selector
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.launch
import java.time.LocalDateTime
import java.util.Optional
import java.util.concurrent.BlockingQueue

class ZenohManager() {

    companion object {
        val TAG: String = ZenohManager::class.java.simpleName
        const val ZENOH_RESPONSE_TOPIC = "trade/response"
        const val ZENOH_HEARTBEAT_TOPIC= "trade/heartbeat"
        //qtrade.rs
        const val TRADE_STRATEGIES_TOPIC = "trade/strategies";
        const val TRADE_ORDERS_TOPIC   = "trade/orders";
        const val TRADE_HOLDINGS_TOPIC = "trade/holdings";


        val heartbeatKeyExpr:KeyExpr? = try{
            KeyExpr.tryFrom(ZENOH_HEARTBEAT_TOPIC)
        }catch (e: ZenohException) {
            Log.e(TAG, "Failed to create KeyExpr ", e)
            null
        }

        val ordersKeyExpr:KeyExpr? = try{
            KeyExpr.tryFrom(TRADE_ORDERS_TOPIC)
        }catch (e: ZenohException) {
            Log.e(TAG, "Failed to create KeyExpr ", e)
            null
        }


        val responseKeyExpr:KeyExpr? = try{
            KeyExpr.tryFrom(ZENOH_RESPONSE_TOPIC)
        }catch (e: ZenohException) {
            Log.e(TAG, "Failed to create KeyExpr ", e)
            null
        }

        val strategiesSelector: Selector? = try{
            Selector.tryFrom(TRADE_STRATEGIES_TOPIC)
        }catch (e: KeyExprException) {
            Log.e(TAG, "Failed to create Selector ", e)
            null
        }

        val ordersSelector: Selector? = try{
            Selector.tryFrom(TRADE_ORDERS_TOPIC)
        }catch (e: KeyExprException) {
            Log.e(TAG, "Failed to create Selector ", e)
            null
        }

        val holdingsSelector:Selector? = try{
            Selector.tryFrom(TRADE_HOLDINGS_TOPIC)
        }catch (e: KeyExprException) {
            Log.e(TAG, "Failed to create Selector ", e)
            null
        }
        private val zenohCoroutineScope = CoroutineScope(Dispatchers.IO)

    }

    private lateinit var session: Session
    val listeners = mutableMapOf<String, MutableList<ZenohListener>>()
    private var isSubscribed = mutableMapOf<String, Boolean>()


    init {
        try {
            session = Session.open()
        } catch (e: Exception) {
            // Handle initialization errors
            Log.e(TAG, "Failed to open Zenoh session", e)
        }
    }


    fun publish(keyExpr: KeyExpr, payload:String){
        val publisher = session.declarePublisher(
            keyExpr
        ).congestionControl(CongestionControl.BLOCK).res()

        publisher.put(payload).res()

        publisher.close()
    }

    fun subscribe(keyExpr: KeyExpr, listener: ZenohListener) {
        val keyString = keyExpr.toString()

        synchronized(listeners) {
            listeners.computeIfAbsent(keyString) { mutableListOf() }.also { listenersList ->
                if (!listenersList.contains(listener)) {
                    listenersList.add(listener)
                }
            }

            if (isSubscribed[keyString] == null ) {
                // has to use CoroutineScope, or else will block ui thread
                zenohCoroutineScope.launch {
                    try {
                        session.declareSubscriber(keyExpr).res().use { subscriber ->
                            val receiver = checkNotNull(subscriber.receiver)
                            while (true) {
                                val wrapper = receiver.take()
                                val sample = wrapper.get()
                                Log.d(TAG,"\n\n")
                                Log.i(TAG, "@@ listeners= $listeners for ${sample.keyExpr}")
                                Log.i(TAG, ">> [ZenohManager@$this] Received " + sample.kind + " ('" + sample.keyExpr + "': '" + sample.value + "')")

                                val sampleKeyString = sample.keyExpr.toString()
                                synchronized(listeners) {
                                    listeners[sampleKeyString]?.forEach { l ->
                                    l.onZenohSampleReceived(sample)
                                    }
                                }
                            }
                        }
                    } catch (e: ZenohException) {
                        Log.e(TAG, "Failed to create KeyExpr for subscription", e)
                    } finally {
                        synchronized(listeners) {
                            isSubscribed[keyString] = false
                        }
                    }
                }
                isSubscribed[keyString] = true
            }
        }
    }

    fun unsubscribe(keyExpr: KeyExpr, listener: ZenohListener) {
        val keyString = keyExpr.toString()
        synchronized(listeners) {
            listeners[keyString]?.remove(listener)
        }
    }

    fun select(selector: Selector,callback: ZenohCallback){
        val now = LocalDateTime.now()
        Log.d(TAG, ">> $now (select) Performing on '$selector'...")

        val receiver: BlockingQueue<Optional<Reply>> = checkNotNull(session.get(selector).res())

        while (true) {
            val now = LocalDateTime.now()
            val wrapper = receiver.take()

            if (wrapper == null) {
                Log.d(TAG, ">> $now (select)  break null")
                break
            }

            val reply = wrapper.orElse(null)

            if (reply == null) {
                Log.d(TAG,">> $now (select) break null")
                break
            }

            when (reply) {
                is Reply.Success -> {
                    val successReply = reply as Reply.Success
                    Log.d(TAG,">>  $now (select) Received ('" + successReply.sample.keyExpr + "': '" + successReply.sample.value + "')")
                    callback.onSuccess(successReply.sample)
                }
                is Reply.Error -> {
                    val errorReply = reply as Reply.Error
                    Log.d(TAG,">> $now (select) Received (ERROR: '" + errorReply.error + "')")
                    callback.onError(errorReply.error.toString())
                }
            }
        }
    }

    /**
     * Subscriber received Sample from Zenoh topic
     */
    fun interface ZenohListener {
        // Subscriber received Sample from Zenoh topic
        fun onZenohSampleReceived(sample: Sample)
    }

    /**
     * Getter received Sample from Zenoh topic
     */
    interface ZenohCallback {
        // Define a callback interface for processing the received data
        fun onSuccess(sample: Sample)
        fun onError(error: String)
    }
}